package State

import ProcessFunctionTest.SplitTempProcessor
import Source.SensorReading
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.runtime.state.memory.MemoryStateBackend
import org.apache.flink.streaming.api.scala._

object 状态后端 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //设置状态后端
    //    env.setStateBackend(new MemoryStateBackend())
    //    env.setStateBackend(new FsStateBackend("file:\\E:\\IDEA\\IntelliJ IDEA 2019.1\\bm\\Flink_Scala\\src\\main\\resources\\state"))
    env.setStateBackend(new RocksDBStateBackend("file:\\E:\\IDEA\\IntelliJ IDEA 2019.1\\bm\\Flink_Scala\\src\\main\\resources\\RockState"))

    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }).keyBy(_.id)

    //分流
    val highTempStream = dataStream
      .process(new SplitTempProcessor(30.0))

    highTempStream.print("high")
    //获取侧输出流，测输出流的输出类型和主流不一定一致,这边的输入需要和下面定义的侧输出流一致
    highTempStream.getSideOutput(new OutputTag[(String, Long, Double)]("low")).print("low")

    env.execute()
  }
}
